package com.xyzwps.reactor;

import reactor.core.publisher.Flux;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class Ex01Flux {
    public static void main(String[] args) throws IOException {
        System.out.println("选择 demo：");
        var reader = new BufferedReader(new InputStreamReader(System.in));
        switch (reader.readLine()) {
            case "just" -> demoJust();
            case "generate" -> demoGenerate();
            case "create" -> demoCreate();
            case "push" -> demoPush();
            default -> System.out.println("啥也没选");
        }
    }

    /**
     * 演示基本使用
     */
    static void demoJust() {
        Flux.just("Hello", "Flux", "World", "ex")
                .map(it -> {
                    if ("ex".equals(it)) throw new RuntimeException("fuck");
                    else return "[" + it + "]";
                })
                .subscribe(System.out::println, Ex01Flux::handleEx, Ex01Flux::handleComplete);
    }

    static void handleEx(Throwable t) {
        System.out.println("Error: " + t.getMessage());
    }

    static void handleComplete() {
        System.out.println("Complete.");
    }

    /**
     * 演示 generate
     */
    static void demoGenerate() {
        Flux.generate(
                () -> 0, // 初始状态
                (state, sink) -> {
                    sink.next("3 × " + state + " = " + 3 * state);
                    if (state == 10) sink.complete();
                    return state + 1;
                }
        ).subscribe(it -> System.out.println("Sub: " + it));
    }

    /**
     * 演示 create: 用于多线程产生状态的环境
     */
    static void demoCreate() {
        Flux.create(sink -> {
            for (int state = 0; state < 5; state++) {
                sink.next("3 × " + state + " = " + 3 * state);
            }
            sink.complete();
        }).subscribe(it -> System.out.println("Sub: " + it));
    }

    /**
     * 演示 push: 用于单线程产生状态的环境
     */
    static void demoPush() {
        Flux.push(sink -> {
            for (int state = 0; state < 5; state++) {
                sink.next("3 × " + state + " = " + 3 * state);
            }
            sink.complete();
        }).subscribe(it -> System.out.println("Sub: " + it));
    }
}
